from pyspark.sql import SparkSession

"""
@version:
@software: PyCharm
@file: 
@time: 
"""

spark = SparkSession.builder.master(master="local").appName("test").getOrCreate()

sc = spark.sparkContext

wordList = ["cat", "elephant", "ret", "rat", "cat", "cat", "cat"]

wordsRDD = sc.parallelize(wordList, 4)


# 方法一
def makePlural(word):
    return word + "s"  # 将传进来的值加s返回


# print(makePlural('cat'))
appliedRDD1 = wordsRDD.map(makePlural)  # 将RDD中的每一个项目执行makePlural方法并返回
# print("在调用collect之后开始执行：{}".format(appliedRDD1.collect()))  # 没有调用之前没有执行

# 方法二
appliedRDD2 = wordsRDD.map(lambda x: x + "s")  # 将wordsRDD中的每一项都加“s”然后返回
print("在调用collect之后开始执行：{}".format(appliedRDD2.collect()))  # 没有调用之前没有执行

# part2：使用pair RDD 来做计算
# pair RDD是一种（key，velue）方式存储的RDD
# [{"":""},{"":""}]

pairRDD = wordsRDD.map(lambda word: (word, 1))
# print("打印pairRDD第一项：{}".format(pairRDD.take(1)))
print("查看所有pairRDD运算结果：{}".format(pairRDD.collect()))

# groupBykey():通过相同的key分组
wordsGrouped = pairRDD.groupByKey()
for key, value in wordsGrouped.collect():
    print("{0}:{1}".format(key, list(value)))

##################################

# 依照key值加总  方法一  测试没有成功，
# wordCountsGrouped = wordsGrouped.map(lambda k: (k1, sum(list(k2))))
# print("查看相同key相加的结果：{}".format(wordCountsGrouped.collect()))

##################################

# 依照key值加总  方法二  测试成功，通过reduceByKey，可以
wordCounts = pairRDD.reduceByKey(lambda a, b: a + b)
print("reduceByKey能够直接将资料根据key值聚合，减少多余的交换(shuffle)动作:{}".format(wordCounts.collect()))

# pyspark中wordcount的小案例

wordsList2 = ['cat', 'elephant', 'rat', 'rat', 'cat']
wordRDD2 = sc.parallelize(wordsList2)
wordCountRDD_demo = (wordRDD2.map(lambda x: (x, 1))  # 将原来的列表转换为（key，value）格式
                     .reduceByKey(lambda x, y: x + y))  # 将转换之后的的rdd按key聚合然后，把value进行相加操作

print("workCount通过两个函数执行:{}".format(wordCountRDD_demo.collect()))
